package com.ksc.wordcount.shuffle.nettyimpl.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

import java.io.FileWriter;

import java.util.HashMap;
import java.util.Map;

public class ShuffleService {

    int serverPort = 0; // 服务器端口号

    String baseDir = "/tmp/"; // 文件存储的基本路径

    Map<Integer, FileWriter> map = new HashMap(); // 存储每个连接的文件写入器

    public ShuffleService(int serverPort) {
        this.serverPort = serverPort;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用于接受客户端的连接请求
        EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用于处理接受到的连接的数据
        ServerBootstrap b = new ServerBootstrap().group(bossGroup, workerGroup); // 创建启动服务的引导类

        try {
            b
                    .channel(NioServerSocketChannel.class) // 设置服务端的通道类型为NioServerSocketChannel
                    .childOption(ChannelOption.SO_KEEPALIVE, true) // 设置是否保持连接
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ObjectEncoder(), // 序列化对象为字节流
                                    new ObjectDecoder(ClassResolvers.cacheDisabled(null)), // 反序列化字节流为对象
                                    new ShuffleServiceHandler()); // 自定义的处理器
                        }
                    });

            ChannelFuture f = b.bind(serverPort).sync(); // 绑定服务器端口，开始监听客户端连接
            System.out.println("Server started at port " + serverPort);
            f.channel().closeFuture().sync(); // 等待服务器端口被关闭
        } finally {
            workerGroup.shutdownGracefully(); // 关闭工作线程
            bossGroup.shutdownGracefully(); // 关闭主线程
        }
    }

}